package com.imooc.springcloud.biz;

import com.imooc.springcloud.topic.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @date 2021-04-02
 * @author zkw
 */
@Slf4j
@EnableBinding(
        //绑定信道
        value = {
                Sink.class,
                MyTopic.class,
                GroupTopic.class,
                DelayedTopic.class,
                ErrorTopic.class,
                RequeueTopic.class,
                DlqTopic.class,
                FallbackTopic.class
        }
)
public class StreamConsumer {

    private AtomicInteger count = new AtomicInteger(1);

    @StreamListener(Sink.INPUT)
    public void consume(Object payload) {
        log.info("message consumed successfully, payload={}", payload);
    }

    @StreamListener(MyTopic.INPUT)
    public void consumeMyMessage(Object payload) {
        log.info("My message consumed successfully, payload={}", payload);
    }

    @StreamListener(GroupTopic.INPUT)
    public void consumeGroupMessage(Object payload) {
        log.info("Group message consumed successfully, payload={}", payload);
    }

    @StreamListener(DelayedTopic.INPUT)
    public void consumeDelayedMessage(MessageBean bean) {
        log.info("Delayed message consumed successfully, payload={}", bean.getPayload());
    }

    /**
     * 异常重试（单机版）
     * @param bean
     */
    @StreamListener(ErrorTopic.INPUT)
    public void consumeErrorMessage(MessageBean bean) {
        log.info("Are you OK? ");
        if (count.incrementAndGet() % 3 == 0) {
            log.info("Yes,I'm fine.");
            count.set(0);
        } else {
            log.info("pardon?  I didn't hear you clearly. Please say it again！ ");
            throw new RuntimeException("I'm not OK.");
        }
        log.info("Error message consumed successfully, payload={}", bean.getPayload());
    }

    /**
     * 异常重试（联机版-重新入列）
     * @param bean
     */
    @StreamListener(RequeueTopic.INPUT)
    public void requeueErrorMessage(MessageBean bean) {
        log.info("Are you OK? ");
        try {
            Thread.sleep(3000);
        } catch (Exception e) {
        }
        //throw new RuntimeException("I'm not OK.");
        log.info("error message consumed successfully, payload={}", bean.getPayload());
    }

    @StreamListener(DlqTopic.INPUT)
    public void consumeDlqMessage(MessageBean bean) {
        log.info("Dlq - Are you OK? ");
        System.out.println("dlq count=" + count);
        int dd = count.incrementAndGet();
        System.out.println("dlq dd=" + dd);
        if (dd % 5 == 0) {
            log.info("Dlq - Yes,I'm fine.");
        } else {
            log.info("Dlq - pardon?  I didn't hear you clearly. Please say it again！ ");
            throw new RuntimeException("Dlq - I'm not OK.");
        }
    }

    /**
     * fallback + 版本升级
     * @param bean
     * @param version
     */
    @StreamListener(FallbackTopic.INPUT)
    public void consumeFallbackMessage(MessageBean bean,
                                       @Header("version") String version) {
        log.info("fallback - Are you OK? ");
        if (ObjectUtils.nullSafeEquals("1.0", version)) {
            log.info("fallback - Yes,I'm fine.");
        } else if (ObjectUtils.nullSafeEquals("2.0", version)) {
            log.info("unsupported version");
            throw new RuntimeException("fallback - I'm not OK.");
        } else {
            log.info("fallback - version={}", version);
        }
    }

    /**
     * 当业务请求抛出异常且重试次数达到上限后，会被转到fallback降级方法
     * @param message
     */
    @ServiceActivator(inputChannel = "fallback-topic.fallback-group.errors")
    public void fallback(Message<?> message) {
        log.info("fallback entered:");
    }

}
